package com.hefan.common.ons;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PropertyValueConst;
import com.google.common.collect.Lists;
import com.hefan.common.ons.listener.GLLMessageListener;
import com.hefan.common.util.DynamicProperties;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ApplicationObjectSupport;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

/**
 * @author: ninglijun
 * @date: 15/5/22
 * @time: 下午12:10
 * @description:
 */
@Component
public class ConsumerFactory extends ApplicationObjectSupport {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerFactory.class);
    private static final List<Consumer> allConsumers = Lists.newArrayList();

    @PostConstruct
    private void loadConsumer() {
        String env = DynamicProperties.getString("ons.env");
        if (StringUtils.equals(env, "local")) {
            loadTestConsumer();
            return;
        }else if (StringUtils.equals(env, "dev")) {
            Map<TopicRegistryDev, GLLMessageListener> listeners = getListenersDev();
            for (String cid : TopicRegistryDev.allConsumerIds()) {
                logger.info("loading consumer : {}", cid);
                int count = 0;
                Consumer consumer = createConsumer(cid);
                for (TopicRegistryDev registry : TopicRegistryDev.getTopicsByConsumerId(cid)) {
                    if (listeners.get(registry) == null) continue;
                    consumer.subscribe(registry.getTopic(), "*", listeners.get(registry));
                    count++;
                }
                if (count > 0) {
                    allConsumers.add(consumer);
                }
            }
        }else if (StringUtils.equals(env, "test")) {
            Map<TopicRegistryTest, GLLMessageListener> listeners = getListenersTest();
            for (String cid : TopicRegistryTest.allConsumerIds()) {
                logger.info("loading consumer : {}", cid);
                int count = 0;
                Consumer consumer = createConsumer(cid);
                for (TopicRegistryTest registry : TopicRegistryTest.getTopicsByConsumerId(cid)) {
                    if (listeners.get(registry) == null) continue;
                    consumer.subscribe(registry.getTopic(), "*", listeners.get(registry));
                    count++;
                }
                if (count > 0) {
                    allConsumers.add(consumer);
                }
            }
        }else if (StringUtils.equals(env, "online")) {
            Map<TopicRegistry, GLLMessageListener> listeners = getListeners();
            for (String cid : TopicRegistry.allConsumerIds()) {
                if (StringUtils.equals(cid, TopicRegistry.HEFANTV_TEST2.getConsumerId())) {
                    continue;
                }
                logger.info("loading consumer : {}", cid);
                int count = 0;
                Consumer consumer = createConsumer(cid);
                for (TopicRegistry registry : TopicRegistry.getTopicsByConsumerId(cid)) {
                    if (listeners.get(registry) == null) continue;
                    consumer.subscribe(registry.getTopic(), "*", listeners.get(registry));
                    count++;
                }
                if (count > 0) {
                    allConsumers.add(consumer);
                }
            }
        }

    }

    private void loadTestConsumer() {
        Map<TopicRegistry, GLLMessageListener> listeners = getListeners();
        Consumer consumer = createConsumer(TopicRegistry.HEFANTV_TEST2.getConsumerId());
        logger.info("loading consumer : {}", TopicRegistry.HEFANTV_TEST2.getConsumerId());
        int count = 0;
        for (TopicRegistry registry : TopicRegistry.getTopicsByConsumerId(TopicRegistry.HEFANTV_TEST2.getConsumerId())) {
            if (listeners.get(registry) == null) continue;
            consumer.subscribe(registry.getTopic(), "*", listeners.get(registry));
            count++;
        }
        if (count > 0) {
            allConsumers.add(consumer);
        }
    }

    public static void startAll() {
        for (Consumer consumer : allConsumers) {
            consumer.start();
            logger.info("consumer : {} stared", consumer.toString());
        }
    }

    private Map<TopicRegistry, GLLMessageListener> getListeners() {
        ApplicationContext ac = getApplicationContext();
        Map<TopicRegistry, GLLMessageListener> result = new HashMap<>();
        Map<String, GLLMessageListener> beanMap = ac.getBeansOfType(GLLMessageListener.class);
        for (Map.Entry<String, GLLMessageListener> entry : beanMap.entrySet()) {
            result.put(entry.getValue().getTopicRegistry(), entry.getValue());
        }
        return result;
    }
    private Map<TopicRegistryDev, GLLMessageListener> getListenersDev() {
        ApplicationContext ac = getApplicationContext();
        Map<TopicRegistryDev, GLLMessageListener> result = new HashMap<>();
        Map<String, GLLMessageListener> beanMap = ac.getBeansOfType(GLLMessageListener.class);
        for (Map.Entry<String, GLLMessageListener> entry : beanMap.entrySet()) {
            result.put(entry.getValue().getTopicRegistryDev(), entry.getValue());
        }
        return result;
    }
    private Map<TopicRegistryTest, GLLMessageListener> getListenersTest() {
        ApplicationContext ac = getApplicationContext();
        Map<TopicRegistryTest, GLLMessageListener> result = new HashMap<>();
        Map<String, GLLMessageListener> beanMap = ac.getBeansOfType(GLLMessageListener.class);
        for (Map.Entry<String, GLLMessageListener> entry : beanMap.entrySet()) {
            result.put(entry.getValue().getTopicRegistryTest(), entry.getValue());
        }
        return result;
    }

    private Consumer createConsumer(String consumerId) {
        Properties prop = new Properties();
        prop.setProperty(PropertyKeyConst.ConsumerId, consumerId);
        prop.setProperty(PropertyKeyConst.AccessKey, ONSConstants.AccessKey);
        prop.setProperty(PropertyKeyConst.SecretKey, ONSConstants.SecretKey);
        String env = DynamicProperties.getString("ons.env");
        if (StringUtils.equals(env, "local")) {
            prop.put(PropertyKeyConst.MessageModel,
                    PropertyValueConst.BROADCASTING);

        }
        Consumer consumer = null;
        try {
            consumer = ONSFactory.createConsumer(prop);
            logger.info("consumer : " + consumerId + " created");
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("consumer : {} created error, error msg : {}", consumer, e.getMessage());
        }

        return consumer;
    }

}
